home *** CD-ROM | disk | FTP | other *** search
/ Complete Linux / Complete Linux.iso / docs / apps / database / postgres / postgre4.z / postgre4 / src / executor / n_hash.c < prev    next >
Encoding:
C/C++ Source or Header  |  1992-08-27  |  26.9 KB  |  1,012 lines
  1. /* ----------------------------------------------------------------
  2.  *   FILE
  3.  *    hash.c
  4.  *    
  5.  *   DESCRIPTION
  6.  *    Routines to hash relations for hashjoin
  7.  *
  8.  *   INTERFACE ROUTINES
  9.  *         ExecHash    - generate an in-memory hash table of the relation 
  10.  *         ExecInitHash    - initialize node and subnodes..
  11.  *         ExecEndHash    - shutdown node and subnodes
  12.  *
  13.  *   NOTES
  14.  *    
  15.  *   IDENTIFICATION
  16.  *    $Header: /private/postgres/src/executor/RCS/n_hash.c,v 1.17 1992/08/19 01:10:30 mer Exp $
  17.  * ----------------------------------------------------------------
  18.  */
  19.  
  20. #include <math.h>
  21. #include <sys/file.h>
  22. #include "storage/ipci.h"
  23. #include "storage/bufmgr.h"    /* for BLCKSZ */
  24. #include "tcop/slaves.h"
  25. #include "executor/executor.h"
  26.  
  27.  RcsId("$Header: /private/postgres/src/executor/RCS/n_hash.c,v 1.17 1992/08/19 01:10:30 mer Exp $");
  28.  
  29.  
  30. extern int NBuffers;
  31. static int HashTBSize;
  32.  
  33. /* ----------------------------------------------------------------
  34.  *       ExecHash
  35.  *
  36.  *    build hash table for hashjoin, all do partitioning if more
  37.  *    than one batches are required.
  38.  * ----------------------------------------------------------------
  39.  */
  40. /**** xxref:
  41.  *           ExecProcNode
  42.  ****/
  43. TupleTableSlot
  44. ExecHash(node)
  45. Hash node;
  46. {
  47.     EState      estate;
  48.     HashState      hashstate;
  49.     Plan       outerNode;
  50.     Var            hashkey;
  51.     HashJoinTable hashtable;
  52.     HeapTuple     heapTuple;
  53.     TupleTableSlot slot;
  54.     ExprContext      econtext;
  55.  
  56.     int          nbatch;
  57.     File      *batches;
  58.     RelativeAddr  *batchPos;
  59.     int          *batchSizes;
  60.     char      *tempname;
  61.     int          i;
  62.     RelativeAddr  *innerbatchNames;
  63.     
  64.     /* ----------------
  65.      *    get state info from node
  66.      * ----------------
  67.      */
  68.     
  69.     hashstate =   get_hashstate(node);
  70.     estate =      (EState) get_state((Plan) node);
  71.     outerNode =   get_outerPlan((Plan) node);
  72.  
  73.     if (!IsMaster && ParallelExecutorEnabled()) {
  74.     IpcMemoryId  shmid;
  75.     IpcMemoryKey hashtablekey;
  76.     int         hashtablesize;
  77.     int         tmpfd;
  78.  
  79.     hashtablekey = get_hashtablekey(node);
  80.     hashtablesize = get_hashtablesize(node);
  81.     
  82.     /* ----------------
  83.      *  in Sequent version, shared memory is implemented by 
  84.      *  memory mapped files, it takes one file descriptor.
  85.      *  we may have to free one for this.
  86.      * ----------------
  87.      */
  88.     closeOneVfd();
  89.         shmid = IpcMemoryCreateWithoutOnExit(hashtablekey,
  90.                             hashtablesize,
  91.                             HASH_PERMISSION);
  92.         hashtable = (HashJoinTable) IpcMemoryAttach(shmid);
  93.     set_hashtable(node, hashtable);
  94.     }
  95.     
  96.     hashtable =    get_hashtable(node);
  97.     if (hashtable == NULL)
  98.     elog(WARN, "ExecHash: hash table is NULL.");
  99.     
  100.     nbatch = hashtable->nbatch;
  101.     
  102.     if (nbatch > 0) {  /* if needs hash partition */
  103.     innerbatchNames = (RelativeAddr *) ABSADDR(hashtable->innerbatchNames);
  104.  
  105.     /* --------------
  106.      *  allocate space for the file descriptors of batch files
  107.      *  then open the batch files in the current processes.
  108.      * --------------
  109.      */
  110.     batches = (File*)palloc(nbatch * sizeof(File));
  111.     for (i=0; i<nbatch; i++) {
  112.         batches[i] = FileNameOpenFile(ABSADDR(innerbatchNames[i]),
  113.                      O_CREAT | O_RDWR, 0600);
  114.     }
  115.     set_hashBatches(hashstate, batches);
  116.         batchPos = (RelativeAddr*) ABSADDR(hashtable->innerbatchPos);
  117.         batchSizes = (int*) ABSADDR(hashtable->innerbatchSizes);
  118.     }
  119.  
  120.     /* ----------------
  121.      *    set expression context
  122.      * ----------------
  123.      */
  124.     hashkey = get_hashkey(node);
  125.     econtext = get_cs_ExprContext((CommonState) hashstate);
  126.  
  127.     /* ----------------
  128.      *    get tuple and insert into the hash table
  129.      * ----------------
  130.      */
  131.     for (;;) {
  132.     slot = ExecProcNode(outerNode);
  133.     if (TupIsNull((Pointer)slot))
  134.         break;
  135.     
  136.     set_ecxt_innertuple(econtext, slot);
  137.     ExecHashTableInsert(hashtable, econtext, hashkey, 
  138.                 get_hashBatches(hashstate));
  139.  
  140.     ExecClearTuple((Pointer) slot);
  141.     }
  142.     
  143.     /*
  144.      * end of build phase, flush all the last pages of the batches.
  145.      */
  146.     for (i=0; i<nbatch; i++) {
  147.     if (FileSeek(batches[i], 0L, L_XTND) < 0)
  148.         perror("FileSeek");
  149.     if (FileWrite(batches[i],ABSADDR(hashtable->batch)+i*BLCKSZ,BLCKSZ) < 0)
  150.         perror("FileWrite");
  151.     NDirectFileWrite++;
  152.    }
  153.     
  154.     /* ---------------------
  155.      *  Return the slot so that we have the tuple descriptor 
  156.      *  when we need to save/restore them.  -Jeff 11 July 1991
  157.      * ---------------------
  158.      */
  159.     return slot;
  160. }
  161.  
  162. /* ----------------------------------------------------------------
  163.  *       ExecInitHash
  164.  *
  165.  *       Init routine for Hash node
  166.  * ----------------------------------------------------------------
  167.  */
  168. /**** xxref:
  169.  *           ExecInitNode
  170.  ****/
  171. List
  172. ExecInitHash(node, estate, parent)
  173.     Hash     node;
  174.     EState     estate;
  175.     Plan    parent;
  176. {
  177.     HashState        hashstate;
  178.     Plan        outerPlan;
  179.     
  180.     SO1_printf("ExecInitHash: %s\n",
  181.            "initializing hash node");
  182.     
  183.     /* ----------------
  184.      *  assign the node's execution state
  185.      * ----------------
  186.      */
  187.     set_state((Plan) node,  (EStatePtr)estate);
  188.     
  189.     /* ----------------
  190.      * create state structure
  191.      * ----------------
  192.      */
  193.     hashstate = MakeHashState(NULL);
  194.     set_hashstate(node, hashstate);
  195.     set_hashBatches(hashstate, NULL);
  196.     
  197.     /* ----------------
  198.      *  Miscellanious initialization
  199.      *
  200.      *         +    assign node's base_id
  201.      *       +    assign debugging hooks and
  202.      *       +    create expression context for node
  203.      * ----------------
  204.      */
  205.     ExecAssignNodeBaseInfo(estate, (BaseNode) hashstate, parent);
  206.     ExecAssignDebugHooks((Plan) node, (BaseNode) hashstate);
  207.     ExecAssignExprContext(estate, (CommonState) hashstate);
  208.     
  209. #define HASH_NSLOTS 1
  210.     /* ----------------
  211.      * initialize our result slot
  212.      * ----------------
  213.      */
  214.     ExecInitResultTupleSlot(estate, (CommonState) hashstate);
  215.     
  216.     /* ----------------
  217.      * initializes child nodes
  218.      * ----------------
  219.      */
  220.     outerPlan = get_outerPlan((Plan) node);
  221.     ExecInitNode(outerPlan, estate, (Plan) node);
  222.     
  223.     /* ----------------
  224.      *     initialize tuple type. no need to initialize projection
  225.      *  info because this node doesn't do projections
  226.      * ----------------
  227.      */
  228.     ExecAssignResultTypeFromOuterPlan((Plan) node, (CommonState) hashstate);
  229.     set_cs_ProjInfo((CommonState) hashstate, NULL);
  230.  
  231.     return
  232.     LispTrue;
  233. }
  234.  
  235. int
  236. ExecCountSlotsHash(node)
  237.     Plan node;
  238. {
  239.     return ExecCountSlotsNode(get_outerPlan(node)) +
  240.        ExecCountSlotsNode(get_innerPlan(node)) +
  241.        HASH_NSLOTS;
  242. }
  243.  
  244. /* ---------------------------------------------------------------
  245.  *       ExecEndHash
  246.  *
  247.  *    clean up routine for Hash node
  248.  * ----------------------------------------------------------------
  249.  */
  250.  
  251. /**** xxref:
  252.  *           ExecEndNode
  253.  ****/
  254. void
  255. ExecEndHash(node)
  256.     Hash node;
  257. {
  258.     HashState        hashstate;
  259.     Plan        outerPlan;
  260.     File        *batches;
  261.     
  262.     /* ----------------
  263.      *    get info from the hash state 
  264.      * ----------------
  265.      */
  266.     hashstate = get_hashstate(node);
  267.     batches = get_hashBatches(hashstate);
  268.     if (batches != NULL)
  269.        pfree(batches);
  270.  
  271.     /* ----------------
  272.      *    free projection info.  no need to free result type info
  273.      *  because that came from the outer plan...
  274.      * ----------------
  275.      */
  276.     ExecFreeProjectionInfo((CommonState) hashstate);
  277.     
  278.     /* ----------------
  279.      *    shut down the subplan
  280.      * ----------------
  281.      */
  282.     outerPlan = get_outerPlan((Plan) node);
  283.     ExecEndNode(outerPlan);
  284. } 
  285.  
  286. RelativeAddr
  287. hashTableAlloc(size, hashtable)
  288. int size;
  289. HashJoinTable hashtable;
  290. {
  291.     RelativeAddr p;
  292.     p = hashtable->top;
  293.     hashtable->top += size;
  294.     return p;
  295. }
  296.  
  297. /* ----------------------------------------------------------------
  298.  *    ExecHashTableCreate
  299.  *
  300.  *    create a hashtable in shared memory for hashjoin.
  301.  * ----------------------------------------------------------------
  302.  */
  303. #define NTUP_PER_BUCKET        10
  304. #define FUDGE_FAC        1.5
  305.  
  306. HashJoinTable
  307. ExecHashTableCreate(node)
  308.     Plan node;
  309. {
  310.     Plan      outerNode;
  311.     int          nbatch;
  312.     int       ntuples;
  313.     int       tupsize;
  314.     IpcMemoryId   shmid;
  315.     HashJoinTable hashtable;
  316.     HashBucket       bucket;
  317.     int       nbuckets;
  318.     int          totalbuckets;
  319.     int       bucketsize;
  320.     int       i;
  321.     int          tmpfd;
  322. #ifdef sequent
  323.     slock_t       *batchLock;
  324. #endif
  325.     RelativeAddr  *outerbatchNames;
  326.     RelativeAddr  *outerbatchPos;
  327.     RelativeAddr  *innerbatchNames;
  328.     RelativeAddr  *innerbatchPos;
  329.     int          *innerbatchSizes;
  330.     RelativeAddr  tempname;
  331.  
  332.     nbatch = -1;
  333.     HashTBSize = NBuffers/2;
  334.     while (nbatch < 0) {
  335.     /*
  336.      * determine number of batches for the hashjoin
  337.      */
  338.     HashTBSize *= 2;
  339.     nbatch = ExecHashPartition((Hash) node);
  340.       }
  341.     /* ----------------
  342.      *    get information about the size of the relation
  343.      * ----------------
  344.      */
  345.     outerNode = get_outerPlan(node);
  346.     ntuples = get_plan_size(outerNode);
  347.     if (ntuples <= 0)
  348.     ntuples = 1000;  /* XXX just a hack */
  349.     tupsize = get_plan_width(outerNode) + sizeof(HeapTupleData);
  350.  
  351.     /*
  352.      * totalbuckets is the total number of hash buckets needed for
  353.      * the entire relation
  354.      */
  355.     totalbuckets = ceil((double)ntuples/NTUP_PER_BUCKET);
  356.     bucketsize = NTUP_PER_BUCKET * tupsize + sizeof(*bucket);
  357.     
  358.     /*
  359.      * nbuckets is the number of hash buckets for the first pass
  360.      * of hybrid hashjoin
  361.      */
  362.     nbuckets = (HashTBSize - nbatch) * BLCKSZ / (bucketsize * FUDGE_FAC);
  363.     if (totalbuckets < nbuckets)
  364.        totalbuckets = nbuckets;
  365.     if (nbatch == 0)
  366.        nbuckets = totalbuckets;
  367. #ifdef HJDEBUG
  368.     printf("nbatch = %d, totalbuckets = %d, nbuckets = %d\n", nbatch, totalbuckets, nbuckets);
  369. #endif
  370.  
  371. #ifdef sequent
  372.     /* ----------------
  373.      *    in Sequent version, shared memory is implemented by 
  374.      *  memory mapped files, it takes one file descriptor.
  375.      *  we may have to free one for this.
  376.      * ----------------
  377.      */
  378.     closeOneVfd();
  379.     if (ParallelExecutorEnabled()) {
  380.        IpcMemoryKey hashtablekey;
  381.        int        hashtablesize;
  382.        hashtablekey = get_hashtablekey((Hash)node);
  383.        hashtablesize = (HashTBSize+SlaveLocalInfoD.nparallel)*BLCKSZ;
  384.        shmid = IpcMemoryCreateWithoutOnExit(hashtablekey,
  385.                             hashtablesize,
  386.                             HASH_PERMISSION);
  387.        set_hashtablesize((Hash)node, hashtablesize);
  388.       }
  389.     else
  390.        shmid = IpcMemoryCreateWithoutOnExit(PrivateIPCKey,
  391.                      (HashTBSize+1)*BLCKSZ,HASH_PERMISSION);
  392.     hashtable = (HashJoinTable) IpcMemoryAttach(shmid);
  393. #else /* sequent */
  394.     /* ----------------
  395.      *  in non-parallel machines, we don't need to put the hash table
  396.      *  in the shared memory.  We just palloc it.
  397.      * ----------------
  398.      */
  399.     hashtable = (HashJoinTable)palloc((HashTBSize+1)*BLCKSZ);
  400. #endif /* sequent */
  401.  
  402.     if (hashtable == NULL) {
  403.     elog(WARN, "not enough memory for hashjoin.");
  404.       }
  405.     /* ----------------
  406.      *    initialize the hash table header
  407.      * ----------------
  408.      */
  409.     hashtable->nbuckets = nbuckets;
  410.     hashtable->totalbuckets = totalbuckets;
  411.     hashtable->bucketsize = bucketsize;
  412.     hashtable->shmid = shmid;
  413.     hashtable->top = sizeof(HashTableData);
  414. #ifdef sequent
  415.     S_INIT_LOCK(&(hashtable->overflowlock));
  416. #endif
  417.     hashtable->bottom = HashTBSize * BLCKSZ;
  418.     /*
  419.      *  hashtable->readbuf has to be long aligned!!!
  420.      */
  421.     hashtable->readbuf = hashtable->bottom;
  422.     hashtable->nbatch = nbatch;
  423.     hashtable->curbatch = 0;
  424.     hashtable->pcount = hashtable->nprocess = SlaveLocalInfoD.nparallel;
  425. #ifdef sequent
  426.     S_INIT_BARRIER(&(hashtable->batchBarrier), hashtable->nprocess);
  427.     S_INIT_LOCK(&(hashtable->tableLock));
  428. #endif
  429.     if (nbatch > 0) {
  430. #ifdef sequent
  431.     /* ---------------
  432.      *  allocate and initialize locks for each batch
  433.      * ---------------
  434.      */
  435.     batchLock = (slock_t*)ABSADDR(
  436.                hashTableAlloc(nbatch * sizeof(slock_t), hashtable));
  437.     for (i=0; i<nbatch; i++) 
  438.         S_INIT_LOCK(&(batchLock[i]));
  439.     hashtable->batchLock = RELADDR(batchLock);
  440. #endif
  441.     /* ---------------
  442.      *  allocate and initialize the outer batches
  443.      * ---------------
  444.      */
  445.     outerbatchNames = (RelativeAddr*)ABSADDR(
  446.         hashTableAlloc(nbatch * sizeof(RelativeAddr), hashtable));
  447.     outerbatchPos = (RelativeAddr*)ABSADDR(
  448.             hashTableAlloc(nbatch * sizeof(RelativeAddr), hashtable));
  449.     for (i=0; i<nbatch; i++) {
  450.         tempname = hashTableAlloc(12, hashtable);
  451.         mk_hj_temp(ABSADDR(tempname));
  452.         outerbatchNames[i] = tempname;
  453.         outerbatchPos[i] = -1;
  454.       }
  455.     hashtable->outerbatchNames = RELADDR(outerbatchNames);
  456.     hashtable->outerbatchPos = RELADDR(outerbatchPos);
  457.     /* ---------------
  458.      *  allocate and initialize the inner batches
  459.      * ---------------
  460.      */
  461.     innerbatchNames = (RelativeAddr*)ABSADDR(
  462.            hashTableAlloc(nbatch * sizeof(RelativeAddr), hashtable));
  463.     innerbatchPos = (RelativeAddr*)ABSADDR(
  464.            hashTableAlloc(nbatch * sizeof(RelativeAddr), hashtable));
  465.     innerbatchSizes = (int*)ABSADDR(
  466.                  hashTableAlloc(nbatch * sizeof(int), hashtable));
  467.     for (i=0; i<nbatch; i++) {
  468.         tempname = hashTableAlloc(12, hashtable);
  469.         mk_hj_temp(ABSADDR(tempname));
  470.         innerbatchNames[i] = tempname;
  471.         innerbatchPos[i] = -1;
  472.         innerbatchSizes[i] = 0;
  473.       }
  474.     hashtable->innerbatchNames = RELADDR(innerbatchNames);
  475.     hashtable->innerbatchPos = RELADDR(innerbatchPos);
  476.     hashtable->innerbatchSizes = RELADDR(innerbatchSizes);
  477.     }
  478.     else {
  479.     hashtable->outerbatchNames = (RelativeAddr)NULL;
  480.     hashtable->outerbatchPos = (RelativeAddr)NULL;
  481.     hashtable->innerbatchNames = (RelativeAddr)NULL;
  482.     hashtable->innerbatchPos = (RelativeAddr)NULL;
  483.     hashtable->innerbatchSizes = (RelativeAddr)NULL;
  484.     }
  485.  
  486.     hashtable->batch = (RelativeAddr)LONGALIGN(hashtable->top + 
  487.                            bucketsize * nbuckets);
  488.     hashtable->overflownext=hashtable->batch + nbatch * BLCKSZ;
  489.     /* ----------------
  490.      *    initialize each hash bucket
  491.      * ----------------
  492.      */
  493.     bucket = (HashBucket)ABSADDR(hashtable->top);
  494.     for (i=0; i<nbuckets; i++) {
  495.     bucket->top = RELADDR((char*)bucket + sizeof(*bucket));
  496.     bucket->bottom = bucket->top;
  497.     bucket->firstotuple = bucket->lastotuple = -1;
  498. #ifdef sequent
  499.     S_INIT_LOCK(&(bucket->bucketlock));
  500. #endif
  501.     bucket = (HashBucket)((char*)bucket + bucketsize);
  502.     }
  503.     return(hashtable);
  504. }
  505.  
  506. /* ----------------------------------------------------------------
  507.  *    ExecHashTableInsert
  508.  *
  509.  *    insert a tuple into the hash table depending on the hash value
  510.  *    it may just go to a tmp file for other batches
  511.  * ----------------------------------------------------------------
  512.  */
  513. void
  514. ExecHashTableInsert(hashtable, econtext, hashkey, batches)
  515.     HashJoinTable hashtable;
  516.     ExprContext   econtext;
  517.     Var       hashkey;
  518.     File      *batches;
  519. {
  520.     TupleTableSlot    slot;
  521.     HeapTuple         heapTuple;
  522.     Const         hashvalue;
  523.     HashBucket         bucket;
  524.     int            bucketno;
  525.     int            nbatch;
  526.     int            batchno;
  527.     char        *buffer;
  528.     RelativeAddr    *batchPos;
  529.     int            *batchSizes;
  530. #ifdef sequent
  531.     slock_t        *batchLock;
  532. #endif
  533.     char        *pos;
  534.  
  535.     nbatch =         hashtable->nbatch;
  536.     batchPos =         (RelativeAddr*)ABSADDR(hashtable->innerbatchPos);
  537.     batchSizes =     (int*)ABSADDR(hashtable->innerbatchSizes);
  538. #ifdef sequent
  539.     batchLock =     (slock_t*)ABSADDR(hashtable->batchLock);
  540. #endif
  541.  
  542.     slot = get_ecxt_innertuple(econtext);
  543.     heapTuple = (HeapTuple) CAR(slot);
  544.     
  545. #ifdef HJDEBUG
  546.     printf("Inserting ");
  547. #endif
  548.     
  549.     bucketno = ExecHashGetBucket(hashtable, econtext, hashkey);
  550.  
  551.     /* ----------------
  552.      *    decide whether to put the tuple in the hash table or a tmp file
  553.      * ----------------
  554.      */
  555.     if (bucketno < hashtable->nbuckets) {
  556.     /* ---------------
  557.      *  put the tuple in hash table
  558.      * ---------------
  559.      */
  560.     bucket = (HashBucket)
  561.         (ABSADDR(hashtable->top) + bucketno * hashtable->bucketsize);
  562. #ifdef sequent
  563.     /* --------------
  564.      *  lock the hash bucket
  565.      * ---------------
  566.      */
  567.         if (ParallelExecutorEnabled())
  568.        S_LOCK(&(bucket->bucketlock));
  569. #endif
  570.     if ((char*)LONGALIGN(ABSADDR(bucket->bottom))
  571.          -(char*)bucket+heapTuple->t_len > hashtable->bucketsize)
  572.         ExecHashOverflowInsert(hashtable, bucket, heapTuple);
  573.     else {
  574.         bcopy(heapTuple,(char*)LONGALIGN(ABSADDR(bucket->bottom)),
  575.           heapTuple->t_len); 
  576.         bucket->bottom = 
  577.         ((RelativeAddr)LONGALIGN(bucket->bottom) + heapTuple->t_len);
  578.     }
  579. #ifdef sequent
  580.     /* --------------
  581.      *  insertion done, unlock the hash bucket
  582.      * --------------
  583.      */
  584.         if (ParallelExecutorEnabled())
  585.        S_UNLOCK(&(bucket->bucketlock));
  586. #endif
  587.     }
  588.     else {
  589.     /* -----------------
  590.      * put the tuple into a tmp file for other batches
  591.      * -----------------
  592.      */
  593.     batchno = (float)(bucketno - hashtable->nbuckets)/
  594.           (float)(hashtable->totalbuckets - hashtable->nbuckets)
  595.           * nbatch;
  596.     buffer = ABSADDR(hashtable->batch) + batchno * BLCKSZ;
  597. #ifdef sequent
  598.     /* ----------------
  599.      *  lock the batch
  600.      * ----------------
  601.      */
  602.     if (ParallelExecutorEnabled())
  603.         S_LOCK(&(batchLock[batchno]));
  604. #endif
  605.     batchSizes[batchno]++;
  606.     pos= (char *)
  607.         ExecHashJoinSaveTuple(heapTuple,
  608.                   buffer,
  609.                   batches[batchno],
  610.                   ABSADDR(batchPos[batchno]));
  611.     batchPos[batchno] = RELADDR(pos);
  612. #ifdef sequent
  613.     /* -----------------
  614.      *  unlock the batch
  615.      * -----------------
  616.      */
  617.     if (ParallelExecutorEnabled())
  618.         S_UNLOCK(&(batchLock[batchno]));
  619. #endif
  620.     }
  621. }
  622.  
  623. /* ----------------------------------------------------------------
  624.  *    ExecHashTableDestroy
  625.  *
  626.  *    destroy a hash table
  627.  * ----------------------------------------------------------------
  628.  */
  629. void
  630. ExecHashTableDestroy(hashtable)
  631.     HashJoinTable hashtable;
  632. {
  633. #ifdef sequent
  634.     IPCPrivateMemoryKill(0, hashtable->shmid);
  635. #else /* sequent */
  636.     pfree((Pointer)hashtable);
  637. #endif /* sequent */
  638. }
  639.  
  640. /* ----------------------------------------------------------------
  641.  *    ExecHashGetBucket
  642.  *
  643.  *    Get the hash value for a tuple
  644.  * ----------------------------------------------------------------
  645.  */
  646. int
  647. ExecHashGetBucket(hashtable, econtext, hashkey)
  648.     HashJoinTable hashtable;
  649.     ExprContext   econtext;
  650.     Var       hashkey;
  651. {
  652.     int     bucketno;
  653.     HashBucket     bucket;
  654.     Const     hashvalue;
  655.     Datum     keyval;
  656.     extern Boolean execConstByVal;
  657.     extern int execConstLen;
  658.     Boolean isNull;
  659.  
  660.  
  661.     /* ----------------
  662.      *    Get the join attribute value of the tuple
  663.      * ----------------
  664.      */
  665.     keyval = ExecEvalVar(hashkey, econtext, &isNull);
  666.     
  667.     /* ------------------
  668.      *  compute the hash function
  669.      * ------------------
  670.      */
  671.     if (execConstByVal)
  672.         bucketno =
  673.         hashFunc((char *) &keyval, execConstLen) % hashtable->totalbuckets;
  674.     else
  675.         bucketno =
  676.         hashFunc((char *) keyval, execConstLen) % hashtable->totalbuckets;
  677. #ifdef HJDEBUG
  678.     if (bucketno >= hashtable->nbuckets)
  679.        printf("hash(%d) = %d SAVED\n", keyval, bucketno);
  680.     else
  681.        printf("hash(%d) = %d\n", keyval, bucketno);
  682. #endif
  683.  
  684.     return(bucketno);
  685. }
  686.  
  687. /* ----------------------------------------------------------------
  688.  *    ExecHashOverflowInsert
  689.  *
  690.  *    insert into the overflow area of a hash bucket
  691.  * ----------------------------------------------------------------
  692.  */
  693. void
  694. ExecHashOverflowInsert(hashtable, bucket, heapTuple)
  695.     HashJoinTable hashtable;
  696.     HashBucket       bucket;
  697.     HeapTuple       heapTuple;
  698. {
  699.     OverflowTuple otuple;
  700.     RelativeAddr  newend;
  701.     OverflowTuple firstotuple;
  702.     OverflowTuple lastotuple;
  703.  
  704.     firstotuple = (OverflowTuple)ABSADDR(bucket->firstotuple);
  705.     lastotuple = (OverflowTuple)ABSADDR(bucket->lastotuple);
  706. #ifdef sequent
  707.     if (ParallelExecutorEnabled()) {
  708.         /* -----------------
  709.          *  Here we assume that the current bucket lock has already
  710.      *  been acquried.  But we still need to lock the overflow lock.
  711.      *  This could be a potential bottleneck.
  712.      * -----------------
  713.      */
  714.     S_LOCK(&(hashtable->overflowlock));
  715.        }
  716. #endif
  717.     /* ----------------
  718.      *    see if we run out of overflow space
  719.      * ----------------
  720.      */
  721.     newend = (RelativeAddr)LONGALIGN(hashtable->overflownext + sizeof(*otuple)
  722.          + heapTuple->t_len);
  723.     if (newend > hashtable->bottom) {
  724. #ifdef sequent
  725.     elog(WARN, "hash table out of memory.");
  726. #else
  727.     elog(DEBUG, "hash table out of memory. expanding.");
  728.     /* ------------------
  729.      * XXX this is a temporary hack
  730.      * eventually, recursive hash partitioning will be 
  731.      * implemented
  732.      * ------------------
  733.      */
  734.     hashtable->readbuf = hashtable->bottom = 2 * hashtable->bottom;
  735.     hashtable = (HashJoinTable)repalloc(hashtable, hashtable->bottom+BLCKSZ);
  736.     if (hashtable == NULL) {
  737.         perror("repalloc");
  738.         elog(WARN, "can't expand hashtable.");
  739.       }
  740. #endif
  741.     }
  742.     
  743.     /* ----------------
  744.      *    establish the overflow chain
  745.      * ----------------
  746.      */
  747.     otuple = (OverflowTuple)ABSADDR(hashtable->overflownext);
  748.     hashtable->overflownext = newend;
  749. #ifdef sequent
  750.     /* ------------------
  751.      *  unlock the overflow chain
  752.      * ------------------
  753.      */
  754.     if (ParallelExecutorEnabled())
  755.     S_UNLOCK(&(hashtable->overflowlock));
  756. #endif
  757.     if (firstotuple == NULL)
  758.     bucket->firstotuple = bucket->lastotuple = RELADDR(otuple);
  759.     else {
  760.     lastotuple->next = RELADDR(otuple);
  761.     bucket->lastotuple = RELADDR(otuple);
  762.      }
  763.     
  764.     /* ----------------
  765.      *    copy the tuple into the overflow area
  766.      * ----------------
  767.      */
  768.     otuple->next = -1;
  769.     otuple->tuple = RELADDR(LONGALIGN(((char*)otuple + sizeof(*otuple))));
  770.     bcopy(heapTuple, ABSADDR(otuple->tuple), heapTuple->t_len);
  771. }
  772.  
  773. /* ----------------------------------------------------------------
  774.  *    ExecScanHashBucket
  775.  *
  776.  *    scan a hash bucket of matches
  777.  * ----------------------------------------------------------------
  778.  */
  779. HeapTuple
  780. ExecScanHashBucket(hjstate, bucket, curtuple, hjclauses, econtext)
  781.     HashJoinState     hjstate;
  782.     HashBucket         bucket;
  783.     HeapTuple         curtuple;
  784.     List         hjclauses;
  785.     ExprContext     econtext;
  786. {
  787.     HeapTuple         heapTuple;
  788.     bool         qualResult;
  789.     OverflowTuple     otuple = NULL;
  790.     OverflowTuple     curotuple;
  791.     TupleTableSlot    inntuple;
  792.     OverflowTuple    firstotuple;
  793.     OverflowTuple    lastotuple;
  794.     HashJoinTable    hashtable;
  795.  
  796.     hashtable = get_hj_HashTable(hjstate);
  797.     firstotuple = (OverflowTuple)ABSADDR(bucket->firstotuple);
  798.     lastotuple = (OverflowTuple)ABSADDR(bucket->lastotuple);
  799.     
  800.     /* ----------------
  801.      *    search the hash bucket
  802.      * ----------------
  803.      */
  804.     if (curtuple == NULL || curtuple < (HeapTuple)ABSADDR(bucket->bottom)) {
  805.         if (curtuple == NULL)
  806.         heapTuple = (HeapTuple)
  807.         LONGALIGN(ABSADDR(bucket->top));
  808.     else 
  809.         heapTuple = (HeapTuple)
  810.         LONGALIGN(((char*)curtuple+curtuple->t_len));
  811.     
  812.         while (heapTuple < (HeapTuple)ABSADDR(bucket->bottom)) {
  813.     
  814.         inntuple = (TupleTableSlot)
  815.         ExecStoreTuple((Pointer) heapTuple,    /* tuple to store */
  816.                    (Pointer) get_hj_HashTupleSlot(hjstate),
  817.                    /* slot */
  818.                    InvalidBuffer,    /* tuple has no buffer */
  819.                    false);        /* do not pfree this tuple */
  820.         
  821.         set_ecxt_innertuple(econtext, inntuple);
  822.         qualResult = ExecQual(hjclauses, econtext);
  823.         
  824.         if (qualResult)
  825.         return heapTuple;
  826.        
  827.         heapTuple = (HeapTuple)
  828.         LONGALIGN(((char*)heapTuple+heapTuple->t_len));
  829.     }
  830.     
  831.     if (firstotuple == NULL)
  832.         return NULL;
  833.     otuple = firstotuple;
  834.     }
  835.  
  836.     /* ----------------
  837.      *    search the overflow area of the hash bucket
  838.      * ----------------
  839.      */
  840.     if (otuple == NULL) {
  841.     curotuple = get_hj_CurOTuple(hjstate);
  842.     otuple = (OverflowTuple)ABSADDR(curotuple->next);
  843.     }
  844.  
  845.     while (otuple != NULL) {
  846.     heapTuple = (HeapTuple)ABSADDR(otuple->tuple);
  847.     
  848.     inntuple = (TupleTableSlot)
  849.         ExecStoreTuple((Pointer) heapTuple,      /* tuple to store */
  850.                (Pointer) get_hj_HashTupleSlot(hjstate), /* slot */
  851.                InvalidBuffer, /* SP?? this tuple has no buffer */
  852.                false);      /* do not pfree this tuple */
  853.  
  854.     set_ecxt_innertuple(econtext, inntuple);
  855.     qualResult = ExecQual(hjclauses, econtext);
  856.     
  857.     if (qualResult) {
  858.         set_hj_CurOTuple(hjstate, otuple);
  859.         return heapTuple;
  860.     }
  861.      
  862.     otuple = (OverflowTuple)ABSADDR(otuple->next);
  863.     }
  864.     
  865.     /* ----------------
  866.      *    no match
  867.      * ----------------
  868.      */
  869.     return NULL;
  870. }
  871.  
  872. /* ----------------------------------------------------------------
  873.  *    hashFunc
  874.  *
  875.  *    the hash function, copied from Margo
  876.  * ----------------------------------------------------------------
  877.  */
  878.  
  879. int
  880. hashFunc(key,len)
  881.     char *key;
  882.     int len;
  883. {
  884.     register unsigned int h;
  885.     register int l;
  886.     register unsigned char *k;
  887.  
  888.     /*
  889.      * If this is a variable length type, then 'k' points
  890.      * to a "struct varlena" and len == -1.
  891.      * NOTE:
  892.      * VARSIZE returns the "real" data length plus the sizeof the
  893.      * "vl_len" attribute of varlena (the length information).
  894.      * 'k' points to the beginning of the varlena struct, so
  895.      * we have to use "VARDATA" to find the beginning of the "real"
  896.      * data.
  897.      */
  898.     if (len == -1) {
  899.     l = VARSIZE(key) - sizeof(long);    /* 'varlena.vl_len' is a long*/
  900.     k = (unsigned char*) VARDATA(key);
  901.     } else {
  902.     l = len;
  903.     k = (unsigned char *) key;
  904.     }
  905.  
  906.     h = 0;
  907.     
  908.     /*
  909.      * Convert string to integer
  910.      */
  911.     while (l--) h = h * PRIME1 ^ (*k++);
  912.     h %= PRIME2;
  913.  
  914.     return (h);
  915. }
  916.  
  917. /* ----------------------------------------------------------------
  918.  *    ExecHashPartition
  919.  *
  920.  *    determine the number of batches needed for a hashjoin
  921.  * ----------------------------------------------------------------
  922.  */
  923.  
  924. int
  925. ExecHashPartition(node)
  926. Hash node;
  927. {
  928.     Plan    outerNode;
  929.     int        b;
  930.     int        pages;
  931.     int        ntuples;
  932.     int        tupsize;
  933.  
  934.     /*
  935.      * get size information for plan node
  936.      */
  937.     outerNode = get_outerPlan((Plan) node);
  938.     ntuples = get_plan_size(outerNode);
  939.     if (ntuples == 0) ntuples = 1000;
  940.     tupsize = get_plan_width(outerNode) + sizeof(HeapTupleData);
  941.     pages = ceil((double)ntuples * tupsize * FUDGE_FAC / BLCKSZ);
  942.  
  943.     /*
  944.      * if amount of buffer space below hashjoin threshold,
  945.      * return negative
  946.      */
  947.     if (ceil(sqrt((double)pages)) > HashTBSize)
  948.        return -1;
  949.     if (pages <= HashTBSize)
  950.        b = 0;  /* fit in memory, no partitioning */
  951.     else
  952.        b = ceil((double)(pages - HashTBSize)/(double)(HashTBSize - 1));
  953.  
  954.     return b;
  955. }
  956.  
  957. /* ----------------------------------------------------------------
  958.  *    ExecHashTableReset
  959.  *
  960.  *    reset hash table header for new batch
  961.  * ----------------------------------------------------------------
  962.  */
  963.  
  964. void
  965. ExecHashTableReset(hashtable, ntuples)
  966. HashJoinTable hashtable;
  967. int ntuples;
  968. {
  969.     int i;
  970.     HashBucket bucket;
  971.  
  972.     hashtable->nbuckets = hashtable->totalbuckets
  973.     = ceil((double)ntuples/NTUP_PER_BUCKET);
  974.  
  975.     hashtable->overflownext = hashtable->top + hashtable->bucketsize *
  976.                   hashtable->nbuckets;
  977.  
  978.     bucket = (HashBucket)ABSADDR(hashtable->top);
  979.     for (i=0; i<hashtable->nbuckets; i++) {
  980.        bucket->top = RELADDR((char*)bucket + sizeof(*bucket));
  981.        bucket->bottom = bucket->top;
  982.        bucket->firstotuple = bucket->lastotuple = -1;
  983. #ifdef sequent
  984.         S_INIT_LOCK(&(bucket->bucketlock));
  985. #endif
  986.        bucket = (HashBucket)((char*)bucket + hashtable->bucketsize);
  987.     }
  988.    hashtable->pcount = hashtable->nprocess;
  989. }
  990.  
  991. static int hjtmpcnt = 0;
  992.  
  993. void
  994. mk_hj_temp(tempname)
  995. char *tempname;
  996. {
  997.     sprintf(tempname, "HJ%d.%d", getpid(), hjtmpcnt);
  998.     hjtmpcnt = (hjtmpcnt + 1) % 1000;
  999. }
  1000.  
  1001. print_buckets(hashtable)
  1002. HashJoinTable hashtable;
  1003. {
  1004.    int i;
  1005.    HashBucket bucket;
  1006.    bucket = (HashBucket)ABSADDR(hashtable->top);
  1007.    for (i=0; i<hashtable->nbuckets; i++) {
  1008.       printf("bucket%d = (top = %d, bottom = %d, firstotuple = %d, lastotuple = %d)\n", i, bucket->top, bucket->bottom, bucket->firstotuple, bucket->lastotuple);
  1009.       bucket = (HashBucket)((char*)bucket + hashtable->bucketsize);
  1010.     }
  1011. }
  1012.